package com.keepc.excel.component;

import com.keepc.excel.domain.DataRecord;
import com.keepc.excel.mapper.DataRecordMapper;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

@Component
@Slf4j
public class DataRecordDisruptor {

    // Disruptor 环形缓冲区的大小，可根据需要调整
    private static final int BUFFER_SIZE = 1024 * 1024;
    // 批量插入的大小，可根据性能测试调整
    private static final int BATCH_SIZE = 1000;
    private final Disruptor<DataRecordEvent> disruptor;
    private final RingBuffer<DataRecordEvent> ringBuffer;

    @Autowired
    private DataRecordMapper dataRecordMapper;

    public DataRecordDisruptor() {
        log.info("初始化 Disruptor");
        // 事件工厂，用于创建 DataRecordEvent 实例
        EventFactory<DataRecordEvent> factory = DataRecordEvent::new;
        // 创建线程工厂，使用默认的线程创建机制
        ThreadFactory threadFactory = Executors.defaultThreadFactory();
        // 初始化 Disruptor，使用多生产者模式和阻塞等待策略
        this.disruptor = new Disruptor<>(factory, BUFFER_SIZE, threadFactory, ProducerType.MULTI, new BlockingWaitStrategy());
        // 为 Disruptor 注册事件处理器
        this.disruptor.handleEventsWith(new DataRecordEventHandler());
        // 启动 Disruptor
        this.ringBuffer = this.disruptor.start();
    }

    // 发布数据到 Disruptor 的方法
    public void publish(List<DataRecord> dataRecords) {
        for (DataRecord dataRecord : dataRecords) {
            // 获取环形缓冲区的下一个可用序列
            long sequence = this.ringBuffer.next();
            try {
                // 获取事件对象并填充数据
                DataRecordEvent event = this.ringBuffer.get(sequence);
                event.setDataRecord(dataRecord);
                // 标记为正常事件
                event.setEnd(false);
            } finally {
                // 发布事件
                this.ringBuffer.publish(sequence);
            }
        }
        // 发布结束标志事件
        long sequence = this.ringBuffer.next();
        try {
            DataRecordEvent event = this.ringBuffer.get(sequence);
            // 无实际数据
            event.setDataRecord(null);
            // 标记为结束事件
            event.setEnd(true);
        } finally {
            this.ringBuffer.publish(sequence);
        }
    }

    // 内部类，作为 Disruptor 的事件对象
    private static class DataRecordEvent {
        private DataRecord dataRecord;
        private boolean isEnd; // 是否是结束事件

        public DataRecord getDataRecord() {
            return this.dataRecord;
        }

        public void setDataRecord(DataRecord dataRecord) {
            this.dataRecord = dataRecord;
        }

        public boolean isEnd() {
            return this.isEnd;
        }

        public void setEnd(boolean isEnd) {
            this.isEnd = isEnd;
        }
    }

    // 事件处理器
    private class DataRecordEventHandler implements EventHandler<DataRecordEvent> {
        private final List<DataRecord> batch = new ArrayList<>();

        @Override
        public void onEvent(DataRecordEvent event, long sequence, boolean endOfBatch) {
            if (event.isEnd()) {
                // 如果是结束事件，处理剩余数据并退出
                if (!this.batch.isEmpty()) {
                    this.insertBatch(this.batch);
                    this.batch.clear();
                }
                return;
            }

            // 添加数据到批量列表
            this.batch.add(event.getDataRecord());

            // 达到批量大小或是批处理结束时执行插入
            if (this.batch.size() >= BATCH_SIZE || endOfBatch) {
                this.insertBatch(this.batch);
                this.batch.clear();
            }
        }

        private void insertBatch(List<DataRecord> dataRecords) {
            try {
                DataRecordDisruptor.this.dataRecordMapper.insertBatch(dataRecords);
            } catch (Exception e) {
                log.error("批量插入数据失败", e);
            }
        }
    }
}
